package org.chen.ctrip.rmq.producer;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.chen.common.ctrip.entity.Product;
import org.chen.ctrip.rmq.config.RocketMQConfig;
import org.chen.ctrip.rmq.domain.AiProcessingMessage;
import org.chen.ctrip.service.ProductService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

@Slf4j
@Service
public class AiProcessingProducerService {
    
    @Autowired
    private DefaultMQProducer aiProcessingProducer;
    
    @Autowired
    private ProductService productService;
    
    /**
     * 发送AI处理消息
     */
    public boolean sendAiProcessingMessage(String productId, String source) {
        return sendAiProcessingMessage(productId, source, 0);
    }
    
    /**
     * 发送AI处理消息（支持延迟）
     */
    public boolean sendAiProcessingMessage(String productId, String source, int delaySeconds) {
        try {
            // 检查商品是否需要AI处理
            if (!needAiProcessing(productId)) {
                log.info("商品{}不需要AI处理，跳过", productId);
                return false;
            }
            
            AiProcessingMessage aiMessage = new AiProcessingMessage(productId, source);
            String messageBody = aiMessage.toJson();
            
            Message message = new Message(
                RocketMQConfig.AI_PROCESSING_TOPIC,
                "AI_PROCESSING", // Tag
                productId, // Keys，用于消息查找和去重
                messageBody.getBytes(StandardCharsets.UTF_8)
            );
            
            // 设置消息属性
            message.putUserProperty("productId", productId);
            message.putUserProperty("source", source);
            message.putUserProperty("traceId", aiMessage.getTraceId());
            
            // 延迟消息（用于重试）
            if (delaySeconds > 0) {
                // RocketMQ延迟级别：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                int delayLevel = calculateDelayLevel(delaySeconds);
                message.setDelayTimeLevel(delayLevel);
                log.info("发送延迟AI处理消息: productId={}, delaySeconds={}, delayLevel={}", 
                    productId, delaySeconds, delayLevel);
            }
            
            SendResult sendResult = aiProcessingProducer.send(message);
            
            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                log.info("AI处理消息发送成功: productId={}, source={}, msgId={}", 
                    productId, source, sendResult.getMsgId());
                return true;
            } else {
                log.error("AI处理消息发送失败: productId={}, status={}", 
                    productId, sendResult.getSendStatus());
                return false;
            }
            
        } catch (Exception e) {
            log.error("发送AI处理消息异常: productId={}", productId, e);
            return false;
        }
    }
    
    /**
     * 批量发送AI处理消息
     */
    public BatchSendResult batchSendAiProcessingMessages(String source) {
        QueryWrapper<Product> wrapper = new QueryWrapper<>();
        wrapper.eq("detail_fetched", true);
        wrapper.and(w -> w.eq("ask_ai", false).or().isNull("ask_ai"));
        wrapper.isNotNull("page_div");
        wrapper.ne("page_div", "");
        wrapper.orderByAsc("create_time");
        
        List<Product> pendingProducts = productService.list(wrapper);
        
        BatchSendResult result = new BatchSendResult();
        result.setTotalCount(pendingProducts.size());
        
        // 分批发送，避免一次性发送过多消息
        int batchSize = 100;
        for (int i = 0; i < pendingProducts.size(); i += batchSize) {
            int endIndex = Math.min(i + batchSize, pendingProducts.size());
            List<Product> batch = pendingProducts.subList(i, endIndex);
            
            List<Message> messages = new ArrayList<>();
            for (Product product : batch) {
                if (needAiProcessing(product.getId())) {
                    AiProcessingMessage aiMessage = new AiProcessingMessage(product.getId(), source);
                    String messageBody = aiMessage.toJson();
                    
                    Message message = new Message(
                        RocketMQConfig.AI_PROCESSING_TOPIC,
                        "AI_PROCESSING",
                        product.getId(),
                        messageBody.getBytes(StandardCharsets.UTF_8)
                    );
                    
                    message.putUserProperty("productId", product.getId());
                    message.putUserProperty("source", source);
                    message.putUserProperty("traceId", aiMessage.getTraceId());
                    
                    messages.add(message);
                }
            }
            
            if (!messages.isEmpty()) {
                try {
                    SendResult sendResult = aiProcessingProducer.send(messages);
                    if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
                        result.addSuccessCount(messages.size());
                        log.info("批量发送AI处理消息成功: batch={}, count={}", i/batchSize + 1, messages.size());
                    } else {
                        result.addFailCount(messages.size());
                        log.error("批量发送AI处理消息失败: batch={}, status={}", 
                            i/batchSize + 1, sendResult.getSendStatus());
                    }
                } catch (Exception e) {
                    result.addFailCount(messages.size());
                    log.error("批量发送AI处理消息异常: batch={}", i/batchSize + 1, e);
                }
            }
            
            // 批次间延迟
            if (endIndex < pendingProducts.size()) {
                try {
                    Thread.sleep(1000); // 1秒延迟
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
        
        log.info("批量发送AI处理消息完成: 总数={}, 成功={}, 失败={}", 
            result.getTotalCount(), result.getSuccessCount(), result.getFailCount());
        
        return result;
    }
    
    /**
     * 发送到死信队列
     */
    public void sendToDLQ(AiProcessingMessage message, String reason) {
        try {
            message.setRetryCount(999); // 标记为最终失败
            String messageBody = message.toJson();
            
            Message dlqMessage = new Message(
                RocketMQConfig.AI_DLQ_TOPIC,
                "DLQ",
                message.getProductId(),
                messageBody.getBytes(StandardCharsets.UTF_8)
            );
            
            dlqMessage.putUserProperty("reason", reason);
            dlqMessage.putUserProperty("originalTraceId", message.getTraceId());
            
            SendResult sendResult = aiProcessingProducer.send(dlqMessage);
            log.error("消息发送到死信队列: productId={}, reason={}, msgId={}", 
                message.getProductId(), reason, sendResult.getMsgId());
                
        } catch (Exception e) {
            log.error("发送到死信队列失败: productId={}", message.getProductId(), e);
        }
    }
    
    /**
     * 检查商品是否需要AI处理
     */
    private boolean needAiProcessing(String productId) {
        Product product = productService.getById(productId);
        if (product == null) {
            log.warn("商品{}不存在", productId);
            return false;
        }
        
        if (Boolean.TRUE.equals(product.getAskAi())) {
            return false;
        }
        
        if (!Boolean.TRUE.equals(product.getDetailFetched())) {
            return false;
        }
        
        if (product.getPageDiv() == null || product.getPageDiv().trim().isEmpty()) {
            return false;
        }
        
        return true;
    }
    
    /**
     * 计算RocketMQ延迟级别
     */
    private int calculateDelayLevel(int delaySeconds) {
        if (delaySeconds <= 1) return 1;      // 1s
        if (delaySeconds <= 5) return 2;      // 5s
        if (delaySeconds <= 10) return 3;     // 10s
        if (delaySeconds <= 30) return 4;     // 30s
        if (delaySeconds <= 60) return 5;     // 1m
        if (delaySeconds <= 120) return 6;    // 2m
        if (delaySeconds <= 180) return 7;    // 3m
        if (delaySeconds <= 240) return 8;    // 4m
        if (delaySeconds <= 300) return 9;    // 5m
        return 10; // 6m
    }
    
    @Data
    public static class BatchSendResult {
        private int totalCount;
        private int successCount;
        private int failCount;
        
        public void addSuccessCount(int count) {
            this.successCount += count;
        }
        
        public void addFailCount(int count) {
            this.failCount += count;
        }
    }
}